package cn.com.utils;


import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.function.BiFunction;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;

/**
 * CompletableFuture 工具类
 *
 * @author LiuGuodong
 * @date 2024/05/27
 */
@Slf4j
public class CompletableFutureUtil {

    // 默认的执行器服务，使用固定线程池
    private static Executor executor;


    static {
        try {
            executor = SpringContextHolder.getBean("taskExecutor", Executor.class);
        } catch (Exception e) {
            ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
            taskExecutor.setThreadNamePrefix("CompletableFutureThreadPool-");
            taskExecutor.setWaitForTasksToCompleteOnShutdown(true);
            taskExecutor.setCorePoolSize(10);
            taskExecutor.setMaxPoolSize(100);
            taskExecutor.setKeepAliveSeconds(300);
            taskExecutor.setAwaitTerminationSeconds(10);
            taskExecutor.setQueueCapacity(1024);
            taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
            //taskExecutor.setTaskDecorator(new ContextCopyingDecorator());
            taskExecutor.initialize();
            executor = taskExecutor;
            log.error("获取ExecutorService失败，使用默认线程池", e);
        }
    }

    /**
     * 提供一个异步任务的方法
     *
     * @param supplier supplier
     * @param <T>      结果类型
     * @return {@link CompletableFuture }<{@link T }>
     */
    public static <T> CompletableFuture<T> supplyAsync(Supplier<T> supplier) {
        return CompletableFuture.supplyAsync(supplier, executor);
    }

    /**
     * 运行一个异步任务的方法
     *
     * @param runnable runnable
     * @return {@link CompletableFuture }<{@link Void }>
     */
    public static CompletableFuture<Void> runAsync(Runnable runnable) {
        return CompletableFuture.runAsync(runnable, executor);
    }

    /**
     * 创建一个已完成的 CompletableFuture
     *
     * @param value value
     * @param <T>   结果类型
     * @return {@link CompletableFuture }<{@link T }>
     */
    public static <T> CompletableFuture<T> completedFuture(T value) {
        return CompletableFuture.completedFuture(value);
    }

    /**
     * 对 CompletableFuture 的结果应用函数的方法
     *
     * @param future future
     * @param fn     fn
     * @param <T>    原始结果类型
     * @param <U>    转换后的结果类型
     * @return {@link CompletableFuture }<{@link U }>
     */
    public static <T, U> CompletableFuture<U> thenApply(CompletableFuture<T> future, Function<? super T, ? extends U> fn) {
        return future.thenApplyAsync(fn, executor);
    }

    /**
     * 消费 CompletableFuture 结果的方法
     *
     * @param future future
     * @param action action
     * @param <T>    结果类型
     * @return {@link CompletableFuture }<{@link Void }>
     */
    public static <T> CompletableFuture<Void> thenAccept(CompletableFuture<T> future, Consumer<? super T> action) {
        return future.thenAcceptAsync(action, executor);
    }

    /**
     * 在 CompletableFuture 完成后运行任务的方法
     *
     * @param future future
     * @param action action
     * @param <T>    结果类型
     * @return {@link CompletableFuture }<{@link Void }>
     */
    public static <T> CompletableFuture<Void> thenRun(CompletableFuture<T> future, Runnable action) {
        return future.thenRunAsync(action, executor);
    }

    /**
     * 组合两个 CompletableFuture 的方法
     *
     * @param future1 future1
     * @param future2 future2
     * @param fn      fn
     * @param <T>     第一个结果类型
     * @param <U>     第二个结果类型
     * @param <V>     最终结果类型
     * @return {@link CompletableFuture }<{@link V }>
     */
    public static <T, U, V> CompletableFuture<V> thenCombine(CompletableFuture<T> future1, CompletableFuture<U> future2, BiFunction<? super T, ? super U, ? extends V> fn) {
        return future1.thenCombineAsync(future2, fn, executor);
    }

    /**
     * 处理 CompletableFuture 异常的方法
     *
     * @param future future
     * @param fn     fn
     * @param <T>    结果类型
     * @return {@link CompletableFuture }<{@link T }>
     */
    public static <T> CompletableFuture<T> exceptionally(CompletableFuture<T> future, Function<Throwable, ? extends T> fn) {
        return future.exceptionally(fn);
    }

    /**
     * 处理结果和异常的方法
     *
     * @param future future
     * @param fn     fn
     * @param <T>    结果类型
     * @return {@link CompletableFuture }<{@link T }>
     */
    public static <T> CompletableFuture<T> handle(CompletableFuture<T> future, BiFunction<? super T, Throwable, ? extends T> fn) {
        return future.handleAsync(fn, executor);
    }

    /**
     * 等待所有 CompletableFuture 完成的方法
     *
     * @param futures futures
     * @return {@link CompletableFuture }<{@link Void }>
     */
    public static CompletableFuture<Void> allOf(CompletableFuture<?>... futures) {
        return CompletableFuture.allOf(futures);
    }

    /**
     * 等待任意一个 CompletableFuture 完成的方法
     *
     * @param futures futures
     * @return {@link CompletableFuture }<{@link Object }>
     */
    public static CompletableFuture<Object> anyOf(CompletableFuture<?>... futures) {
        return CompletableFuture.anyOf(futures);
    }

    /**
     * 关闭执行器服务
     */
    public static void shutdown() {
        ((ThreadPoolTaskExecutor) executor).shutdown();
    }
}
